[fix][client] Apply no-memory-limit producer queue defaults at producer creation by lhotari · Pull Request #26342 · apache/pulsar

lhotari · 2026-08-17T10:46:01Z

Motivation

The client memory limit is a producer's primary backpressure: it bounds the memory held by messages that have been queued but not yet acknowledged by the broker. #15723 added a safety net for clients that disable it, so that producers fall back to a bounded pending-message queue instead of buffering without any limit. Its commit message states the intent exactly:

restore maxPendingMessages and maxPendingMessagesAcrossPartitions when memory limit is disabled so that pre-PIP-120 default configuration is restored when limit is disabled

PIP-120 (#13344) is what changed those two defaults from 1000/50000 to 0 and made the client memory limit the primary mechanism. So 1000/50000 are defaults — a value the application did not configure — and an application that configures a limit has to keep winning over them, including with an explicit 0, which PIP-120 documented as "disable the pending messages check".

That net was applied on the builder:

public <T> ProducerBuilder<T> newProducer(Schema<T> schema) {
    ProducerBuilderImpl<T> producerBuilder = new ProducerBuilderImpl<>(this, schema);
    if (!memoryLimitController.isMemoryLimited()) {
        producerBuilder.maxPendingMessages(NO_MEMORY_LIMIT_DEFAULT_MAX_PENDING_MESSAGES);
        producerBuilder.maxPendingMessagesAcrossPartitions(
                NO_MEMORY_LIMIT_DEFAULT_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS);
    }
    return producerBuilder;
}

and it has three holes, the first two of which leave a producer with no bound at all, because ProducerImpl only creates its semaphore if (conf.getMaxPendingMessages() > 0):

It is only applied by the newProducer(Schema) overload. The no-argument newProducer() returns a plain builder, so it never gets the defaults. In-tree users of that overload are the WebSocket proxy's producer handler, the Functions log appender and the Functions worker — all on clients whose memory limit is disabled, so all unbounded today.
It is applied by calling the setters on the builder, so a caller that later passes a limit through — say a CLI flag or a config value that happens to be at its own default — silently replaces it. Telling that apart from a deliberate maxPendingMessages(0) is the point of this PR, see below.
Seeding maxPendingMessages to 1000 makes a later maxPendingMessagesAcrossPartitions(500) throw IllegalArgumentException, because that setter requires its value to be >= maxPendingMessages. This is reachable in production: ProducerBuilderFactory builds from newProducer(schema) on a Functions client that is hard-wired to memoryLimit(0), and sets maxPendingMessagesAcrossPartitions independently of maxPendingMessages. A function configured with only maxPendingMessagesAcrossPartitions: 500 fails to create its producer.

This came up while investigating #26340, where pulsar-perf exhausts direct memory against a slow broker. On branch-4.0 its producer hits holes 1 and 2 at once: it calls the no-argument newProducer() and then passes maxPendingMessages(0) through from its own default. #26341 fixes that tool's --memory-limit default; this PR closes the client-side holes that let the safety net be bypassed in the first place.

This PR does not close #26340 on its own. pulsar-perf on the maintenance branches still passes its own unset --max-outstanding / --max-outstanding-across-partitions defaults straight to the builder, which under these semantics reads as an explicit "no limit", and on a non-partitioned topic nothing else bounds it. That needs the same treatment #15283 gave maxPendingMessagesAcrossPartitions — an unset value the tool can recognise, rather than a 0 that means two things — and it is a pulsar-testclient change on branch-4.2/branch-4.0 only: #25887 removed those options' effect on master when pulsar-perf moved to the V5 client. Filed separately.

Modifications

The unset value of maxPendingMessages is 0, and 0 is also the documented value for "no message-count limit". So "never configured" and "explicitly unbounded" cannot be told apart from the configuration alone, and the fix is to record which of the two it is:

ProducerBuilderImpl remembers whether the application called maxPendingMessages(...) / maxPendingMessagesAcrossPartitions(...), or passed either of them to loadConf. The state is carried through clone().
At producer creation the builder asks the client to fill in only the limits that were never configured, on a copy of the configuration, so a default cannot leak into the next producer built from the same builder. A limit that is already positive counts as configured however the configuration was populated, so a configuration built directly is never overwritten either.
newProducer(Schema) no longer seeds the builder. Resolving at creation instead means the defaults no longer depend on which newProducer overload produced the builder (hole 1), cannot be replaced by a pass-through (hole 2), and cannot make a later setter call throw (hole 3).

The result is that the two limits behave as ordinary defaults. On a client whose memory limit is disabled:

configured	`maxPendingMessages`	`maxPendingMessagesAcrossPartitions`
nothing	1000	50000
`maxPendingMessages(500)`	500	50000
`maxPendingMessagesAcrossPartitions(500)`	500	500
`maxPendingMessagesAcrossPartitions(0)`	1000	0
`maxPendingMessages(0)`	0	0

The last row is how an application asks for a producer with no backpressure at all, which is what the previous revision of this PR took away. An explicit maxPendingMessages(0) also leaves the across-partitions budget alone, so that one call is enough whatever the topic's shape: filling the budget in would put a per-partition limit back, because PartitionedProducerImpl derives it from that budget.

Separately, and in its own commit because it is an independent defect: setMaxPendingMessagesAcrossPartitions rejected a value below maxPendingMessages, which made the two setters order-dependent and made ProducerBuilder.loadConf fail outright for any positive maxPendingMessages — ConfigurationDataUtils.loadData replays every property through the setters in an order the caller does not control, so loadConf(Map.of("maxPendingMessages", 5000)) throws maxPendingMessagesAcrossPartitions needs to be >= maxPendingMessages on master today. The check is now >= 0; the relationship is enforced where it is used, in PartitionedProducerImpl, which already lowers the per-partition limit to its share of the budget. This is what ProducerBuilderImplTest#testLoadConfWithAPositiveMaxPendingMessages pins, and it is why loadConf can be treated as a way to configure these limits at all. Happy to split it out if you would rather review it separately.

Two things this deliberately does not do:

The V5 client is unchanged. It reaches producer creation through createSegmentProducerAsync and exposes no pending-message setting at all, so its client memory limit is the only backpressure it has — and therefore the only thing an application can turn off. Filling in a message-count default there would leave a V5 user with no way to express "unbounded". Note that pulsar-perf on master uses the V5 client, so this change on its own does not alter its behaviour there.
Tracking is on the builder, not on ProducerConfigurationData. ProducerBuilder.loadConf goes through ConfigurationDataUtils.loadData, which serialises the configuration to JSON and deserialises a new object through the public setters. A primitive is always emitted, so setMaxPendingMessages(0) always runs on the way back in: a @JsonIgnore marker on the configuration would be set on every loadConf call, and a serialised one would be applied in HashMap key order. The builder is the only place that sees what the application actually called.

ProducerBuilder's javadoc documents what the defaults are and how to opt out. Nothing outside pulsar-client changes.

The WebSocket proxy is left alone. Its maxPendingMessages query parameter is remote-supplied, so an explicit 0 there asks for an unbounded pending queue inside the shared proxy — but the proxy's real problem is that webSocketPulsarClientMemoryLimitInMB is an int defaulting to 0, so it always overrides the client's own 64M default with "no limit". Bounding it by bytes is the better fit and a change of its own, tracked in #26373.

Noticed but deliberately left alone

AbstractReplicator passes the broker's replicationProducerQueueSize straight to maxPendingMessages, so setting that config to 0 leaves a replicator producer unbounded — the replication client has no memory limit either. That is the behaviour on master today and this PR does not change it; a value of 0 is much more likely a misconfiguration than a request for an unbounded queue, but the config is operator-set and undocumented for 0, so tightening it belongs in its own change.
PartitionedProducerImpl writes the per-partition share back into the configuration object it was handed (conf.setMaxPendingMessages(...)). When no default is filled in, that object is the builder's own, so creating a partitioned producer lowers the limit of the next producer built from the same builder. That is pre-existing and unrelated to the memory limit; the fallback path is not affected here because it resolves on a copy.
When maxPendingMessagesAcrossPartitions is explicitly set below the topic's partition count, the per-partition share in PartitionedProducerImpl rounds down to 0, which means "no limit". So a tighter budget produces a looser bound, and an explicitly configured maxPendingMessages is silently discarded. I tried clamping that share to at least 1 and reverted it: the branch it lives in is entered whenever the value is set, with no memory-limit condition, so the clamp also changes behaviour for memory-limited clients — the default configuration. There, an across-partitions budget smaller than the partition count currently yields no semaphore and a working, bytes-bounded producer; clamping turns it into a single permit, and with the default blockIfQueueFull=false the second concurrent in-flight message per partition fails with ProducerQueueIsFullError. That is a much larger blast radius than this PR should carry.

Verifying this change

Make sure that the change passes the CI checks.

This change added tests and can be verified as follows.

Pinning the holes (each was confirmed to fail before the fix):

ProducerQueueSizeTest#testNoArgNewProducerIsBoundedWhenMemoryLimitDisabled — hole 1
ProducerQueueSizeTest#testPartitionedProducerIsBoundedWhenMemoryLimitDisabled
ProducerQueueSizeTest#testFallbackLeavesTheBuilderReusable — pins that the resolved configuration is a copy, and that a later maxPendingMessagesAcrossPartitions call is validated against what the caller configured rather than against a filled-in default (hole 3)
ProducerQueueSizeTest#testExplicitMaxPendingMessagesAboveTheFallbackDoesNotFailCreation and #testExplicitAcrossPartitionsLimitCapsTheFallback — pin that filling in a default never fails producer creation

Pinning that the defaults stay defaults. Each of these fails if "unset" is inferred from the value rather than from what the application configured — verified by reverting that one condition, which fails exactly these:

ProducerQueueSizeTest#testExplicitZeroDisablesTheBoundWhenMemoryLimitDisabled (partitioned and non-partitioned)
ProducerQueueSizeTest#testExplicitZeroAcrossPartitionsKeepsThePerProducerDefault
ProducerQueueSizeTest#testLoadConfZeroDisablesTheBoundWhenMemoryLimitDisabled
ProducerQueueSizeTest#testCloneKeepsAnExplicitlyDisabledBound

Plus ProducerQueueSizeTest#testLoadConfWithoutTheLimitsKeepsTheDefaults (rebuilding the configuration is not mistaken for configuring it), #testMemoryLimitedClientKeepsUnboundedPendingMessages (a client with a memory limit is unaffected).

For the setter fix: ProducerBuilderImplTest#testLoadConfWithAPositiveMaxPendingMessages (fails on master) and #testAcrossPartitionsLimitBelowMaxPendingMessagesIsAccepted.

Also run locally: the whole org.apache.pulsar.client.impl suite, pulsar-functions-instance, pulsar-websocket, ProducerSemaphoreTest, ProducerMemoryLimitTest, MemoryLimitTest, ConsumerMemoryLimitTest, and ./gradlew quickCheck.

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

A producer created on a client whose memory limit is disabled, and which does not configure a pending-message limit, now has a bounded pending-message queue where it previously had none. This is the behaviour #15723 intended; only the cases where it did not take effect change:

The WebSocket proxy is affected out of the box, since webSocketPulsarClientMemoryLimitInMB defaults to 0 and its producers use the no-argument newProducer() with blockIfQueueFull false. A client with more than 1000 unacknowledged messages in flight now gets a failed ProducerAck instead of the proxy buffering them. Raising the limit through the maxPendingMessages query parameter currently also requires enabling batching; the proxy's backpressure is tracked separately in [improve][ws] WebSocket proxy producers have no backpressure: the client memory limit is always disabled #26373.
The Functions log appender and the Functions worker's exclusive producer are likewise bounded now.
On a partitioned topic, an application that set maxPendingMessages but left maxPendingMessagesAcrossPartitions unset now has the filled-in budget divided between the partitions, which can lower its per-partition limit. This matches what the newProducer(Schema) overload already did.
maxPendingMessagesAcrossPartitions no longer throws IllegalArgumentException when it is set below maxPendingMessages; it is accepted, and the per-partition limit is lowered to its share as before. Only code that relied on the exception is affected, and such a call could not previously succeed.

An application that passed maxPendingMessages(0) explicitly to mean "unbounded" keeps that behaviour.

Documentation

doc-required
doc-not-needed
doc
doc-complete

The behaviour is documented in the ProducerBuilder javadoc.

…er creation ### Motivation The client memory limit is a producer's primary backpressure: it bounds the memory held by messages that have been queued but not yet acknowledged by the broker. apache#15723 added a safety net for clients that disable it, so that producers fall back to a bounded pending-message queue instead of buffering without any limit: ```java public <T> ProducerBuilder<T> newProducer(Schema<T> schema) { ProducerBuilderImpl<T> producerBuilder = new ProducerBuilderImpl<>(this, schema); if (!memoryLimitController.isMemoryLimited()) { producerBuilder.maxPendingMessages(NO_MEMORY_LIMIT_DEFAULT_MAX_PENDING_MESSAGES); producerBuilder.maxPendingMessagesAcrossPartitions( NO_MEMORY_LIMIT_DEFAULT_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS); } return producerBuilder; } ``` That net has two holes, and either one leaves a producer with no bound at all, because `ProducerImpl` only creates its semaphore `if (conf.getMaxPendingMessages() > 0)`: 1. It is only applied by the `newProducer(Schema)` overload. The no-argument `newProducer()` returns a plain builder, so it never gets the fallback. In-tree users of that overload include the Functions log appender and the WebSocket proxy's producer handler. 2. It is applied when the builder is constructed, so a later `maxPendingMessages(0)` overwrites it. Since `ProducerConfigurationData.DEFAULT_MAX_PENDING_MESSAGES` is 0, code that passes the default through explicitly silently disables the fallback rather than keeping it. ### Modifications - Resolve the pending-message limits at producer creation, in `PulsarClientImpl`'s `createProducerAsync(conf, schema, interceptors)`, instead of on the builder. That is the funnel every producer built from this client passes through, so the fallback no longer depends on which `newProducer` overload created the builder, and cannot be undone by a later call setting a limit back to its unset value. Only unset limits are filled in; an explicit limit is never overwritten. - Cap the per-producer fallback by the across-partitions limit. That limit is a budget shared by every partition, and its setter rejects a value below `maxPendingMessages`, so filling in the larger default first would throw `IllegalArgumentException` synchronously out of a method that returns a `CompletableFuture`. - Resolve on a copy of the configuration, so filling in a limit does not leak into the next producer built from the same builder. - Remove the now-redundant block from `newProducer(Schema)`. This also fixes a side effect it had: on a client with the memory limit disabled, `newProducer(schema).maxPendingMessagesAcrossPartitions(500)` used to throw, because the builder had already been given a `maxPendingMessages` of 1000. - Document on `ProducerBuilder` what disabling either check actually means. The V5 client reaches producer creation through `createSegmentProducerAsync` and is deliberately left unchanged here; it exposes no pending-message setting of its own and needs a separate decision. Note that `pulsar-perf` on master uses the V5 client, so this change on its own does not alter its behaviour. ### Verifying this change Added tests, each confirmed to fail before the fix: - `ProducerQueueSizeTest#testNoArgNewProducerIsBoundedWhenMemoryLimitDisabled` (hole 1) - `ProducerQueueSizeTest#testLateZeroMaxPendingMessagesDoesNotDisableTheBoundWhenMemoryLimitDisabled` (hole 2) - `ProducerQueueSizeTest#testPartitionedProducerIsBoundedWhenMemoryLimitDisabled` - `ProducerQueueSizeTest#testExplicitMaxPendingMessagesAboveTheFallbackDoesNotFailCreation` and `#testExplicitAcrossPartitionsLimitCapsTheFallback`, which pin that filling in a default never fails producer creation - `ProducerQueueSizeTest#testFallbackLeavesTheBuilderReusable`, which pins that the resolved configuration is a copy `ProducerQueueSizeTest#testMemoryLimitedClientKeepsUnboundedPendingMessages` pins that a client with a memory limit configured is unaffected. Noticed while working on this, left alone as a separate concern: when `maxPendingMessagesAcrossPartitions` is explicitly set below the topic's partition count, the per-partition share in `PartitionedProducerImpl` rounds down to 0, which means "no limit". So a tighter budget produces a looser bound, and an explicitly configured `maxPendingMessages` is silently discarded. On a client with the memory limit disabled it can also defeat the fallback applied here, since the filled-in limit is divided by that same code: an explicit budget of 500 on a topic with 501 partitions still ends up unbounded. It affects clients regardless of their memory limit, and clamping the share turned out to change behaviour for memory-limited clients too, so it needs its own change rather than riding along here. ### Does this pull request potentially affect one of the following parts: - [x] The default values of configurations A producer created on a client whose memory limit is disabled now has a bounded pending-message queue where it previously had none. This is the behaviour apache#15723 intended; only the cases where it did not take effect change. Specifically: - Because `maxPendingMessages` is a primitive `int` whose unset value is 0, an application that explicitly passed 0 to mean "unbounded" cannot be distinguished from one that never set it, and now gets the bound as well. Such an application can keep an unbounded message count by configuring a client memory limit, which bounds the queue by bytes instead, or by setting an explicit `maxPendingMessages`. - On a partitioned topic, an application that set `maxPendingMessages` but left `maxPendingMessagesAcrossPartitions` unset now has the filled-in budget divided between the partitions, which can lower its per-partition limit. This matches what the `newProducer(Schema)` overload already did. - The WebSocket proxy is affected out of the box, since `webSocketPulsarClientMemoryLimitInMB` defaults to 0 and its producers use the no-argument `newProducer()` with `blockIfQueueFull` false. A client with more than 1000 unacknowledged messages in flight now gets a failed `ProducerAck` instead of the proxy buffering them. Raising the limit through the `maxPendingMessages` query parameter currently also requires enabling batching, which is worth fixing separately.

nodece

LGTM

nodece · 2026-08-18T09:15:03Z

Now, the producer has backpressure, how to disable backpressure?

lhotari · 2026-08-18T10:20:30Z

Now, the producer has backpressure, how to disable backpressure?

good question. I'll check and revisit that. The NO_MEMORY_LIMIT_DEFAULT_MAX_PENDING_MESSAGES/NO_MEMORY_LIMIT_DEFAULT_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS solution should only get applied as a template for the producer builder. It might be necessary to expand the PR to cover the cli tools so that it's possible to detect that a parameter hasn't been explicitly set (using Integer instead of int).

…an maxPendingMessages ### Motivation `ProducerConfigurationData.setMaxPendingMessagesAcrossPartitions` rejected any value below `maxPendingMessages`. That makes the two setters order-dependent, and it makes `ProducerBuilder.loadConf` fail outright for any positive `maxPendingMessages`: `ConfigurationDataUtils.loadData` serialises the configuration, merges the caller's map and deserialises a new instance by replaying every property through the public setters, in an order the caller does not control. So builder.loadConf(Map.of("maxPendingMessages", 5000)) throws `maxPendingMessagesAcrossPartitions needs to be >= maxPendingMessages`, because the across-partitions property that comes along with the merged map is still at its default of 0. The same check also makes a builder reject a legitimate call sequence: setting a per-producer limit and then a smaller shared budget throws, while the reverse order is accepted. ### Modifications Validate only that the value is not negative. The relationship between the two limits is enforced where it is used: `PartitionedProducerImpl` lowers the per-partition limit to its share of the budget whenever a budget is set, and the budget is meaningless on a non-partitioned topic. Assisted-by: Claude Code (Opus 5)

…where unset ### Motivation When the client memory limit is disabled there is no byte-based backpressure left, so apache#15723 gave producers the pre-PIP-120 pending-message defaults instead of letting them buffer without any limit. Its commit message states the intent: "restore maxPendingMessages and maxPendingMessagesAcrossPartitions when memory limit is disabled so that pre-PIP-120 default configuration is restored when limit is disabled". They are defaults, and PIP-120 is the same commit that changed them to 0 and documented 0 as "disable the pending messages check". An application that configures a limit therefore has to keep winning over them, including with an explicit 0. The previous approach seeded the defaults onto the builder in `newProducer(Schema)`, which left three holes: 1. The no-argument `newProducer()` never got them, so the WebSocket proxy's producer handler, the Functions log appender and the Functions worker are unbounded today. 2. A caller that later passed a limit through - a CLI flag or a config value sitting at its own default - silently replaced them. 3. Seeding `maxPendingMessages` to 1000 made a later `maxPendingMessagesAcrossPartitions(500)` throw, which a Functions `ProducerConfig` setting only that limit walks straight into. ### Modifications `maxPendingMessages` is a primitive whose unset value is 0, and 0 is also a meaningful explicit value, so the configuration alone cannot tell "never configured" from "explicitly unbounded". `ProducerBuilderImpl` records which of the two limits the application configured - through the setters or through `loadConf` - and carries that across `clone()`. At producer creation the client fills in only the limits that were never configured, on a copy of the configuration so nothing leaks into the next producer built from the same builder. A limit that is already positive counts as configured however the configuration was populated. Tracking this on the builder rather than on the configuration is not a preference: `loadConf` goes through `ConfigurationDataUtils.loadData`, which rebuilds the configuration by replaying every property through the setters, so a marker held there would be re-set on every call. An explicit `maxPendingMessages(0)` also suppresses the across-partitions default, so one call is enough to ask for a producer with no message-count limit whatever the topic's shape - filling in the budget would put a per-partition limit straight back. The V5 client is deliberately left out. It reaches producer creation through `createSegmentProducerAsync` and exposes no pending-message setting at all, so its client memory limit is the only backpressure it has, and the only thing an application can turn off. The WebSocket proxy takes `maxPendingMessages` from a query parameter and its client has no memory limit by default, so a remote client could now ask for an unbounded pending queue inside the shared proxy. A non-positive value is ignored there. Assisted-by: Claude Code (Opus 5)

lhotari · 2026-08-18T12:40:11Z

Now, the producer has backpressure, how to disable backpressure?

You were right — I've reworked the PR so that it can be disabled again.

Why the first revision got this wrong. #15723 introduced the 1000/50000 values to "restore ... pre-PIP-120 default configuration ... when [the memory] limit is disabled" (563a7cb), and PIP-120 (11bfc0e) is the commit that both changed those defaults to 0 and documented 0 as "disable the pending messages check". So they are defaults, and an explicit 0 has to keep meaning "no message-count limit". The first revision turned them into a floor, because it inferred "unset" from maxPendingMessages == 0 — which is the same value as the documented opt-out. With a primitive int there is nothing in ProducerConfigurationData that tells the two apart.

What it does now. ProducerBuilderImpl records whether the application actually called maxPendingMessages(...) / maxPendingMessagesAcrossPartitions(...), or passed either to loadConf, and only a limit that was never configured gets a default filled in — at producer creation, on a copy of the configuration. On a client whose memory limit is disabled:

configured	`maxPendingMessages`	`maxPendingMessagesAcrossPartitions`
nothing	1000	50000
`maxPendingMessages(500)`	500	50000
`maxPendingMessagesAcrossPartitions(500)`	500	500
`maxPendingMessagesAcrossPartitions(0)`	1000	0
`maxPendingMessages(0)`	0	0

So, to disable backpressure entirely — one call for each of the two mechanisms:

PulsarClient client = PulsarClient.builder()
        .serviceUrl(url)
        .memoryLimit(0, SizeUnit.BYTES)   // no byte-based backpressure
        .build();

Producer<byte[]> producer = client.newProducer()
        .topic(topic)
        .maxPendingMessages(0)            // no message-count backpressure
        .create();

An explicit maxPendingMessages(0) deliberately leaves the across-partitions budget alone as well, so that single call is enough on a partitioned topic too — filling the budget in would put a per-partition limit straight back, since PartitionedProducerImpl derives it from that budget. ProducerQueueSizeTest#testExplicitZeroDisablesTheBoundWhenMemoryLimitDisabled runs it for both topic shapes.

Why the flag is on the builder and not on the configuration. I tried the configuration first. ProducerBuilder.loadConf goes through ConfigurationDataUtils.loadData, which serialises the configuration to JSON and deserialises a new object through the public setters. A primitive is always emitted, so setMaxPendingMessages(0) always runs on the way back in: a @JsonIgnore marker would be set on every loadConf call, and a serialised one would be applied in HashMap key order. Changing the fields to Integer is worse — getMaxPendingMessages() would change return type with no possible bridge method, so anything compiled against the current jar gets a NoSuchMethodError, and the class is Serializable. The builder is the only place that sees what the application actually called.

One bug found on the way, in its own commit. setMaxPendingMessagesAcrossPartitions rejected a value below maxPendingMessages. Because loadData replays every property through the setters in an order the caller does not control, that makes loadConf(Map.of("maxPendingMessages", 5000)) throw maxPendingMessagesAcrossPartitions needs to be >= maxPendingMessages on master today — loadConf cannot set a positive maxPendingMessages at all. The check is now >= 0; the relationship is enforced where it is used, in PartitionedProducerImpl. Happy to split that out if you would rather review it separately.

The CLI tools. I thought this PR would have to cover them, but on master they are all on the V5 client now: --max-outstanding / -p no longer reach a v4 producer at all. What does still bite there is --memory-limit: PerformanceBaseArguments.memoryLimit is a long with no initializer and PerfClientUtils passes it through unconditionally, so pulsar-perf disables the client memory limit unless the flag is given — which is exactly #26341. The Integer-instead-of-int change is needed for the branch-4.x backports of this PR, where PerformanceProducer calls the no-argument newProducer() and passes maxOutstanding = DEFAULT_MAX_PENDING_MESSAGES (0) straight through; under these semantics that reads as an explicit "unbounded". I'll make those options Integer in the backport.

V5 is deliberately untouched, and the reason is now in the code: it exposes no pending-message setting at all, so the client memory limit is both its only backpressure and the only thing an application can turn off. Filling in a message-count default there would leave a V5 user with no way to say "unbounded" at all.

One thing added beyond the client: the WebSocket proxy takes maxPendingMessages from a query parameter and its client has no memory limit by default, so now that an explicit 0 is honoured, ?batchingEnabled=true&maxPendingMessages=0 would let a remote client ask for an unbounded pending queue inside the shared proxy. A non-positive value is now ignored.

I've updated the PR description with the full behaviour table and the compatibility notes. Since the mechanism changed, could you take another look?

nodece · 2026-08-18T14:05:31Z

@lhotari LGTM, thanks

The guard on the proxy's remote-supplied maxPendingMessages query parameter belongs with giving the proxy a memory limit, which is a change of its own: `webSocketPulsarClientMemoryLimitInMB` is an `int` defaulting to 0, so the proxy always overrides the client's own 64M default with "no limit". Bounding the proxy by bytes is the better fit there - it multiplexes many producers over one client - and it makes the query parameter a separate question rather than the only defence. Assisted-by: Claude Code (Opus 5)

lhotari · 2026-08-18T18:59:08Z

Thanks @nodece. One change since your review: I reverted the pulsar-websocket part, so the only production code this PR touches is now the client (pulsar-client / pulsar-client-api); the rest is the tests that cover it.

The guard I had added there ignored a non-positive maxPendingMessages query parameter, but that treats the symptom. The proxy's actual problem is that webSocketPulsarClientMemoryLimitInMB is an int defaulting to 0, so the proxy always overrides the client's own 64M default with "no limit" — a memory limit is the right bound for it anyway, since it multiplexes many producers over one client. That is #26373, and the query parameter question belongs with it.

The PR description is updated accordingly.

lhotari requested review from Technoboy-, dao-jun, merlimat and nodece August 17, 2026 10:46

lhotari added release/4.0.4 release/4.2.5 labels Aug 17, 2026

lhotari requested a review from david-streamlio August 17, 2026 10:46

lhotari added release/4.0.14 and removed release/4.0.4 labels Aug 17, 2026

nodece approved these changes Aug 18, 2026

View reviewed changes

lhotari marked this pull request as draft August 18, 2026 10:13

lhotari added 2 commits August 18, 2026 15:37

lhotari marked this pull request as ready for review August 18, 2026 12:39

lhotari requested a review from nodece August 18, 2026 12:40

lhotari mentioned this pull request Aug 18, 2026

[fix][cli][branch-4.2] Don't pass unset pulsar-perf pending-message options to the producer #26371

Open

15 tasks

nodece approved these changes Aug 18, 2026

View reviewed changes

lhotari mentioned this pull request Aug 18, 2026

[improve][ws] WebSocket proxy producers have no backpressure: the client memory limit is always disabled #26373

Open

2 tasks

lhotari merged commit b660d70 into apache:master Aug 18, 2026
44 checks passed

Navigation Menu

Uh oh!

[fix][client] Apply no-memory-limit producer queue defaults at producer creation - #26342

[fix][client] Apply no-memory-limit producer queue defaults at producer creation#26342
lhotari merged 4 commits into
apache:masterfrom
lhotari:lh-fix-nomemorylimit-producer-defaults

lhotari commented Aug 17, 2026 •

edited

Loading

Uh oh!

nodece left a comment

Uh oh!

nodece commented Aug 18, 2026

Uh oh!

lhotari commented Aug 18, 2026

Uh oh!

lhotari commented Aug 18, 2026

Uh oh!

nodece commented Aug 18, 2026

Uh oh!

lhotari commented Aug 18, 2026 •

edited

Loading

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Uh oh!

Conversation

lhotari commented Aug 17, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Motivation

Modifications

Noticed but deliberately left alone

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Uh oh!

nodece left a comment

Choose a reason for hiding this comment

Uh oh!

nodece commented Aug 18, 2026

Uh oh!

lhotari commented Aug 18, 2026

Uh oh!

lhotari commented Aug 18, 2026

Uh oh!

nodece commented Aug 18, 2026

Uh oh!

lhotari commented Aug 18, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

lhotari commented Aug 17, 2026 •

edited

Loading

lhotari commented Aug 18, 2026 •

edited

Loading